#include <stddef.h>
#include <stdint.h>
#include <string.h>
#include <time.h>
#include <algorithm>
#include <map>
#include <iostream>
#include <sstream>

#include "raptor-lite/raptor-lite.h"

const char *g_file_path = NULL;
int g_count             = 0;
struct echo_message {
    uint32_t payload_size;
    uint32_t request_id;
};

struct DownloadEventNode {
    raptor::MultiProducerSingleConsumerQueue::Node node;
    raptor::Endpoint ep;
    FILE *fd;
    int file_size;
    int send_size;
};

class FileServer : public raptor::AcceptorHandler,
                   public raptor::MessageHandler,
                   public raptor::ProtocolHandler,
                   public raptor::EndpointClosedHandler {
public:
    FileServer(/* args */);
    ~FileServer();

    void OnAccept(raptor::Endpoint ep, raptor::Property &settings) {
        settings({{"SocketRecvTimeoutMs", 5000}, {"SocketSendTimeoutMs", 5000}});
        log_debug("OnAccept: %s, fd:%d", ep.PeerString().c_str(), ep.SocketFd());
        container->AttachEndpoint(ep);
    }

    int OnCheckPackageLength(raptor::Endpoint ep, const void *data, size_t len) {
        // log_debug("OnCheckPackageLength:%u", len);
        (void)ep;
        if (len < 4) return 0;
        auto pack_len = *reinterpret_cast<const uint32_t *>(data);
        return static_cast<int>(pack_len) + 8;
    }

    void OnMessage(raptor::Endpoint ep, const raptor::Slice &msg) {
        auto request = reinterpret_cast<const echo_message *>(msg.begin());
        // log_debug("OnMessage: req=%lu millsecs=%lld", request->request_id, request->millseconds);
        auto n = new DownloadEventNode;
        n->ep  = ep;
        n->fd  = NULL;
        mpscq_.push(&n->node);
        count_.FetchAdd(1, raptor::MemoryOrder::RELAXED);
        cv_.Signal();
        log_debug("OnMessage: req=%lu", request->request_id);
    }

    void OnClosed(raptor::Endpoint ep, raptor_error desc) {
        if (desc == RAPTOR_ERROR_NONE) {
            log_debug("OnClosed: %s", ep.PeerString().c_str());
        } else {
            log_debug("OnClosed: %s %s", ep.PeerString().c_str(), desc->ToString().c_str());
        }
    }

    void init() {

        engine = RaptorCreateEngine();

        raptor::Property p{
            {raptor::config::kAcceptorHandler, static_cast<raptor::AcceptorHandler *>(this)},
            {raptor::config::kMessageHandler, static_cast<raptor::MessageHandler *>(this)},
            {raptor::config::kProtocolHandler, static_cast<raptor::ProtocolHandler *>(this)},
            {raptor::config::kEndpointClosedHandler,
             static_cast<raptor::EndpointClosedHandler *>(this)},
            {raptor::config::kCheckReceivingInterval, false},
            {raptor::config::kRecvSendThreads, 1}};

        raptor_error err = engine->CreateContainer(p, &container);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create container: %s", err->ToString().c_str());
            abort();
        }
        err = engine->CreateAcceptor(p, &acceptor);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create acceptor: %s", err->ToString().c_str());
            abort();
        }

        thd_ = raptor::Thread(
            "abc", std::bind(&FileServer::WorkThread, this, std::placeholders::_1), NULL);

        shutdown_ = false;
        count_.Store(0);
    }

    void start_and_listening(const std::string &addr) {
        raptor_error err = acceptor->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to start acceptor: %s", err->ToString().c_str());
            return;
        }
        err = container->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to start container: %s", err->ToString().c_str());
            return;
        }
        err = acceptor->AddListening(addr);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to listening %s, error: %s", addr.c_str(), err->ToString().c_str());
            return;
        }
        thd_.Start();
    }

    void WorkThread(void *) {
        while (!shutdown_) {
            if (count_.Load() < 1) {
                cv_.Wait(&mtx_, 1000);
                continue;
            }
            auto n = mpscq_.pop();
            if (n != nullptr) {
                count_.FetchSub(1, raptor::MemoryOrder::RELAXED);
                auto msg = reinterpret_cast<DownloadEventNode *>(n);
                FileDownload(msg);
                delete msg;
            }
        }
    }
    void stop() {
        acceptor->Shutdown();
        container->Shutdown();
        shutdown_ = true;
        thd_.Join();
    }

    void FileDownload(DownloadEventNode *msg) {
        if (msg->fd == NULL) {
            DownloadEventNode *n = new DownloadEventNode;

            n->fd = fopen(g_file_path, "rb");
            fseek(n->fd, 0, SEEK_END);
            n->file_size = ftell(n->fd);
            fseek(n->fd, 0, SEEK_SET);
            n->send_size = 0;

            n->ep = msg->ep;
            Next(n);
            return;
        }

        if (msg->send_size < msg->file_size) {
            int remain    = msg->file_size - msg->send_size;
            int need_size = RAPTOR_MIN(32768, remain);

            char buff[32768 + 8] = {0};
            int read_bytes       = (int)fread(buff + 8, 1, need_size, msg->fd);
            if (read_bytes > 0) {
                auto hdr          = reinterpret_cast<echo_message *>(buff);
                hdr->payload_size = read_bytes;
                hdr->request_id   = g_count++;
                msg->ep.SendMsg(buff, read_bytes + 8);
            } else {
                log_error("Failed to read file, total: %d, %d", msg->file_size, msg->send_size);
                return;
            }

            DownloadEventNode *n = new DownloadEventNode;
            n->ep                = msg->ep;
            n->fd                = msg->fd;
            n->file_size         = msg->file_size;
            n->send_size         = msg->send_size + read_bytes;
            log_debug("read file, total: %d, %d, %d", read_bytes, n->send_size, n->file_size);

            Next(n);
            return;
        }

        struct echo_message obj;
        obj.payload_size = 0;
        obj.request_id   = 9527;
        msg->ep.SendMsg(&obj, sizeof(obj));
        log_info("Send File Finished");
    }

    void Next(DownloadEventNode *n) {
        mpscq_.push(&n->node);
        count_.FetchAdd(1, raptor::MemoryOrder::RELAXED);
        cv_.Signal();
    }

private:
    bool shutdown_;
    RaptorEngine engine;
    raptor::Container *container = nullptr;
    raptor::Acceptor *acceptor   = nullptr;
    raptor::MultiProducerSingleConsumerQueue mpscq_;
    raptor::Mutex mtx_;
    raptor::Thread thd_;
    raptor::ConditionVariable cv_;
    raptor::AtomicInt32 count_;
};

FileServer::FileServer(/* args */) {
    // Do not call the raptor::Create function here, because the current
    // class (inherited from XXHandler) has not been constructed.
    shutdown_ = true;
}

FileServer::~FileServer() {
    engine->DestroyAcceptor(acceptor);
    engine->DestroyContainer(container);
}

int main(int argc, char *argv[]) {
    if (argc != 2) {
        printf(" Usage: file_server <file_path>\n");
        return 0;
    }
    g_file_path = argv[1];

    printf(" ---- prepare file: %s ---- \n", g_file_path);
    RaptorGlobalStartup();
    FileServer server;
    server.init();
    server.start_and_listening("localhost:50051");
    log_warn("Press any key to quit");
    getchar();
    server.stop();
    RaptorGlobalCleanup();
    return 0;
}
